package com.kl.ordermq;

import com.alibaba.fastjson.JSONObject;
import com.kl.domain.UserDTO;
import com.kl.entity.SeckillCoupon;
import com.kl.entity.SeckillOrder;
import com.kl.service.ISeckillCouponService;
import com.kl.util.IdWorker;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Map;

@Component
@RocketMQMessageListener(topic = "seckill_topic",consumerGroup = "seckill_group")
public class orderMQ implements RocketMQListener, RocketMQPushConsumerLifecycleListener {

    @Resource
    private RedisTemplate redisTemplate;

    @Resource
    private ISeckillCouponService iSeckillCouponService;

    @Resource
    private RocketMQTemplate rocketMQTemplate;



    @Override
    public void onMessage(Object o) {}

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {

        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //获取的消息
                MessageExt messageExt = list.get(0);
                //消息的重试次数
                int times = messageExt.getReconsumeTimes();
                try {
                    //消息的id
                    String msgId = messageExt.getMsgId();
                    //消息幂等性
                    Boolean b = redisTemplate.boundHashOps("SO_MQmessage").hasKey(msgId);
                    if (!b){
                        //消息内容
                        byte[] body = messageExt.getBody();
                        //转成string
                        String message = new String(body);
                        System.out.println(message);
                        //转成map
                        Map map = JSONObject.parseObject(message, Map.class);
                        System.out.println(map);

                        //消费成功放入redis
                        redisTemplate.boundHashOps("SO_MQmessage").put(msgId,messageExt);
                        //消息的主题
                        String topic = messageExt.getTopic();

                        //业务逻辑 根据id查询
                        SeckillCoupon seckillID = iSeckillCouponService.getById(Long.parseLong(map.get("seckillID").toString()));

                        //放入订单表 订单号
                        IdWorker idWorker = new IdWorker();
                        long nextId = idWorker.nextId();

                        UserDTO userDTO = (UserDTO)map.get("user");

                        SeckillOrder order = new SeckillOrder();
                        order.setId(nextId);
                        order.setPaymentStatus(0);
                        order.setPrice(seckillID.getCouponPrice());
                        order.setSeckillCouponId(seckillID.getSeckillCouponId());
                        order.setOrderCreationtime(new Date());
                        order.setOrderUpdatetime(new Date());
                        order.setUserId(Long.parseLong(map.get("userid").toString()));

                        //order.setConsigneeName(userDTO.getName());//收件人姓名
                        //order.setConsigneePhone(userDTO.getPhone());
                        //order.setConsigneeAddress(userDTO.getDefaultAddress());//地址
                        boolean save = order.insert();


                        //发送延时消息 查看用户支付状态
                        rocketMQTemplate.syncSend("sync_topic",MessageBuilder.withPayload(nextId).build(),2000,10);

                        //商品订单放入redis
                        redisTemplate.boundHashOps("seckkill_order").put(map.get("userid"),nextId);


                    }
                }catch (Exception e){
                    //消费失败 判断重试次数 3次之后返回消费成功
                    e.printStackTrace();
                    if (times > 3){
                        //放入自定义死信队列
                        redisTemplate.boundHashOps("SO_DeadLetter").put(messageExt.getMsgId(),messageExt);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                //消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
